package com.rrd.dw.mr.all;

import java.io.IOException;
import java.io.InputStream;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.filter.LongComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HBaseUtil;

/**
 * 全量同步通话详单相关数据，作为kafka数据进行etl的初始化数据
 * @author liubaoxin
 *
 */
public class ExportAllCallRecordsToDW extends Configured implements Tool {
	public static final String CRT001 = "\001";
	public static final Pattern CRT001_PATTERN = Pattern.compile("\001");
	
	 public static final String[] USERINFO_COLUMNS = { "real_name",
		        "in_net_date:date", "level", "email", "address", "zip_code", "mobile",
		        "star_score", "star_level", "status", "gender", "cert_type",
		        "cert_number", "carrier_code", "province", "balance", "data_source" };
		    public static final String[] DETAILS_COLUMNS = { "start_time:date", "area",
		        "mode:int", "to_mobile", "mobile", "duration:int", "type",
		        "total_fee:int", "carrier_code", "normal_fee:int", "roaming_fee:int" };
		    public static final String[] NETFLOWS_COLUMNS = { "nf_start_time:date",
		        "nf_area", "nf_net_type", "nf_duration:int", "nf_flow_type",
		        "nf_total_flows:float", "nf_total_fee:int", "nf_mobile", "nf_source" };
		    public static final String[] SMSBILLS_COLUMNS = { "sms_start_time:date",
		        "sms_area", "sms_to_mobile", "sms_mode:int", "sms_info_type:int",
		        "sms_business_name", "sms_total_fee:int", "sms_mobile" };
		    public static final String[] JOB_COLUMNS = { "job_id", "finished:boolean",
		        "requested:boolean", "successful:boolean", "start_time:date",
		        "end_time:date", "data_file", "system_id", "query", "tryTimes:int",
		        "type", "message", "code", "timestamp:long" };


    @Override
    public int run(String[] args) throws Exception {
        Properties prop = new Properties();
        try (InputStream is = this.getClass().getClassLoader()
            .getResourceAsStream("hbase.properties");) {
            prop.load(is);
        }

        Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
        Configuration conf = job.getConfiguration();
        conf.set("hbase.zookeeper.quorum",
            prop.get("hbase.zookeeper.quorum").toString());
        conf.set("hbase.rootdir", prop.get("hbase.rootdir").toString());

        conf.set("mapreduce.map.memory.mb", "3072");
        conf.set("mapreduce.reduce.memory.mb", "5120");
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec",
            "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs",
            "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("mapreduce.task.io.sort.factor", "10");
        conf.set("hbase.client.scanner.timeout.period", "60000");
        String[] otherArgs = new GenericOptionsParser(conf, args)
            .getRemainingArgs();
        if (otherArgs.length != 4) {
            System.err
                .println("Usage: DataExportTool <out> <start_time> <end_time>");
            return -1;
        }
        String isAll=otherArgs[3];
        System.out.println("isAll==>"+isAll);
        Long startTime = DateUtil.addDay(HBaseUtil.parseDate(otherArgs[1]), -2).getTime();
        Long endTime = DateUtil.addDay(HBaseUtil.parseDate(otherArgs[2]), 2).getTime();
        Long startTime2 = HBaseUtil.parseDate(otherArgs[1]).getTime();
        Long endTime2 = HBaseUtil.parseDate(otherArgs[2]).getTime();
        System.out.println("time_ranage=>"+startTime+","+endTime);
        System.out.println("end_time_filter=>"+startTime2+","+endTime2);
        
        job.setJarByClass(ExportAllCallRecordsToDW.class);
        List<Scan> scans = new ArrayList<Scan>();

        Scan scan = new Scan();
        SingleColumnValueFilter scvf = new SingleColumnValueFilter(Bytes.toBytes("job")
            , Bytes.toBytes("end_time"), CompareFilter.CompareOp.GREATER_OR_EQUAL, 
            new LongComparator(startTime2));
        
        scvf.setFilterIfMissing(true);
        SingleColumnValueFilter scvf2 = new SingleColumnValueFilter(Bytes.toBytes("job")
            , Bytes.toBytes("end_time"), CompareFilter.CompareOp.LESS, 
            new LongComparator(endTime2));
        scvf2.setFilterIfMissing(true);
        List<Filter> filters = new ArrayList<Filter>();
        filters.add(scvf);
//        filters.add(scvf2);
        FilterList filterList=new FilterList(Operator.MUST_PASS_ALL,filters);
        if(isAll.equals("ALL")){
        	System.out.println("进行全量同步！！！！！！！");
        }else{
        	System.out.println("进行增量同步！！！！！！！");
        	scan.setFilter(filterList);
//        	scan.setTimeRange(startTime, endTime + 14400000);
        }
        scan.setCacheBlocks(false);
        
        scan.setCaching(200);
        scan.setMaxVersions(1);
        scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME,
            Bytes.toBytes("call_records"));
       
        
        scans.add(scan);
       
        
        TableMapReduceUtil.initTableMapperJob(scans,
            ExportCallRecordsToDWMapper.class, Text.class, Text.class, job);
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[0]));
        MultipleOutputs.addNamedOutput(job, "userinfo", TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, "details", TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, "netflows", TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, "smsbills", TextOutputFormat.class,
            Text.class, Text.class);
        job.setNumReduceTasks(0);
        int stat= job.waitForCompletion(true) ? 0 : 1;
        System.out.println("成功次數："+job.getCounters().getGroup("call").findCounter("counter").getValue());
        return stat;
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        int res = ToolRunner.run(conf, new ExportAllCallRecordsToDW(), args);
        System.exit(res);
    }

    public static class ExportCallRecordsToDWMapper
            extends TableMapper<Text, Text> {
        private MultipleOutputs<Text, Text> mos;
        private SimpleDateFormat formatter = new SimpleDateFormat(
            "yyyy-MM-dd HH:mm:ss");
        private NullWritable nullWritable = NullWritable.get();
        private String job_id=null;
        private String timestampstr=null;
        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
            this.mos = new MultipleOutputs<Text, Text>(context);
        }

        @Override
        protected void cleanup(Context context)
                throws IOException, InterruptedException {
            this.mos.close();
        }

        @Override
        protected void map(ImmutableBytesWritable key, Result value,
                Context context) throws IOException, InterruptedException {
        	Thread.sleep(5);
            StringBuffer sb = new StringBuffer("");
            Object[] rowkey = HBaseUtil.decodeRowkey(key.get());
            if (rowkey == null) {
                return;
            }
            if(value.getValue(Bytes.toBytes("job"), Bytes.toBytes("job_id"))==null){
            	return;
            }
            long timestamp = Long.parseLong(rowkey[0].toString());
            timestampstr=DateUtil.formateDateStr(new Date(timestamp));
            context.getCounter("records","map.records").increment(1);
            String idNumber = rowkey[1].toString();
            Cell dataCell = null;
            Cell countCell = null;
            String type = "";
            String field = "";
            sb.append(timestampstr).append(CRT001).append(idNumber).append(CRT001);
            //用户基础信息
            for (String str1 : USERINFO_COLUMNS) {
                String[] fieldArray = str1.split(":");
                if (fieldArray.length > 1) {
                    field = fieldArray[0];
                    type = fieldArray[1];
                } else {
                    field = fieldArray[0];
                    type = "string";
                }
                dataCell = value.getColumnLatestCell(Bytes.toBytes("inf"),
                    Bytes.toBytes(field));
                if (dataCell != null) {
                    sb.append(
                        this.getValue(type, CellUtil.cloneValue(dataCell)));
                }
                sb.append(CRT001);
            }
            job_id=Bytes.toString(value.getValue(Bytes.toBytes("job"), Bytes.toBytes("job_id")));
            //job信息
            for (String str1 : JOB_COLUMNS) {
                String[] fieldArray = str1.split(":");
                if (fieldArray.length > 1) {
                    field = fieldArray[0];
                    type = fieldArray[1];
                } else {
                    field = fieldArray[0];
                    type = "string";
                }
                dataCell = value.getColumnLatestCell(Bytes.toBytes("job"),
                    Bytes.toBytes(field));
                if (dataCell != null) {
                    sb.append(
                        this.getValue(type, CellUtil.cloneValue(dataCell)));
                }
                sb.append(CRT001);
            }
            String outString = sb.toString();
            this.mos.write("userinfo", nullWritable,
                new Text(outString.substring(0, outString.lastIndexOf(CRT001))));
            sb = null;
            //通话详单
            countCell = value.getColumnLatestCell(Bytes.toBytes("rec"),
                Bytes.toBytes("count"));
            if (countCell != null) {
                int count = Bytes.toInt(CellUtil.cloneValue(countCell));
                sb = new StringBuffer("");
               
                for (int i = 0; i < count; i++) {
                    int j = 0;
                    sb.append(job_id).append(CRT001);
                    sb.append(timestampstr).append(CRT001).append(idNumber).append(CRT001);
                    for (String str1 : DETAILS_COLUMNS) {
                        String[] fieldArray = str1.split(":");
                        if (fieldArray.length > 1) {
                            field = fieldArray[0];
                            type = fieldArray[1];
                        } else {
                            field = fieldArray[0];
                            type = "string";
                        }
                        dataCell = value.getColumnLatestCell(
                            Bytes.toBytes("rec"),
                            Bytes.toBytes(field + "#" + i));
                        if (dataCell != null) {
                            sb.append(this.getValue(type,
                                CellUtil.cloneValue(dataCell)));
                        }
                        if (j < DETAILS_COLUMNS.length - 1) {
                            sb.append(CRT001);
                        }
                        j++;
                    }
                    if (i < count - 1) {
                        sb.append("\n");
                    }
                }
                this.mos.write("details", nullWritable,
                    new Text(sb.toString()));
            }
            sb = null;
            //流量详单
            countCell = value.getColumnLatestCell(Bytes.toBytes("oth"),
                Bytes.toBytes("nf_count"));
            if (countCell != null) {
                int count = Bytes.toInt(CellUtil.cloneValue(countCell));
                sb = new StringBuffer("");
                for (int i = 0; i < count; i++) {
                    int j = 0;
                    sb.append(job_id).append(CRT001);
                    sb.append(timestampstr).append(CRT001).append(idNumber).append(CRT001);
                    for (String str1 : NETFLOWS_COLUMNS) {
                        String[] fieldArray = str1.split(":");
                        if (fieldArray.length > 1) {
                            field = fieldArray[0];
                            type = fieldArray[1];
                        } else {
                            field = fieldArray[0];
                            type = "string";
                        }
                        dataCell = value.getColumnLatestCell(
                            Bytes.toBytes("oth"),
                            Bytes.toBytes(field + "#" + i));
                        if (dataCell != null) {
                            sb.append(this.getValue(type,
                                CellUtil.cloneValue(dataCell)));
                        }
                        if (j < NETFLOWS_COLUMNS.length - 1) {
                            sb.append(CRT001);
                        }
                        j++;
                    }
                    if (i < count - 1) {
                        sb.append("\n");
                    }
                }
                this.mos.write("netflows",
                    nullWritable,
                    new Text(sb.toString()));
            }
            sb = null;
            //短信详单
            countCell = value.getColumnLatestCell(Bytes.toBytes("oth"),
                Bytes.toBytes("sms_count"));
            if (countCell != null) {
                int count = Bytes.toInt(CellUtil.cloneValue(countCell));
                sb = new StringBuffer("");
                for (int i = 0; i < count; i++) {
                    int j = 0;
                    sb.append(job_id).append(CRT001);
                    sb.append(timestampstr).append(CRT001).append(idNumber).append(CRT001);
                    for (String str1 : SMSBILLS_COLUMNS) {
                        String[] fieldArray = str1.split(":");
                        if (fieldArray.length > 1) {
                            field = fieldArray[0];
                            type = fieldArray[1];
                        } else {
                            field = fieldArray[0];
                            type = "string";
                        }
                        dataCell = value.getColumnLatestCell(
                            Bytes.toBytes("oth"),
                            Bytes.toBytes(field + "#" + i));
                        if (dataCell != null) {
                            sb.append(this.getValue(type,
                                CellUtil.cloneValue(dataCell)));
                        }
                        if (j < SMSBILLS_COLUMNS.length - 1) {
                            sb.append(CRT001);
                        }
                        j++;
                    }
                    if (i < count - 1) {
                        sb.append("\n");
                    }
                }
                this.mos.write("smsbills",
                    nullWritable,
                    new Text(sb.toString()));
            }

        }

        private Object getValue(String type, byte[] value) {
            switch (type) {
                case "string":
                    return value == null ? "" : Bytes.toString(value);
                case "date":
                    return value == null ? new Date(0)
                        : formatter.format(new Date(Bytes.toLong(value)));
                case "bigdecimal":
                    return value == null ? new BigDecimal(0)
                        : Bytes.toBigDecimal(value);
                case "boolean":
                    return value != null && Bytes.toBoolean(value);
                case "double":
                    return value == null ? 0 : Bytes.toDouble(value);
                case "float":
                    return value == null ? 0 : Bytes.toFloat(value);
                case "int":
                    return value == null ? 0 : Bytes.toInt(value);
                case "short":
                    return value == null ? 0 : Bytes.toShort(value);
                case "long":
                    return value == null ? 0 : Bytes.toLong(value);
                default:
                    return Bytes.toString(value);

            }
        }
    }
}
